package com.wwt.friend;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class StepTwo {
    public static class MyMapper2 extends Mapper<LongWritable,Text,Text,Text> {
        @Override
        protected  void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
            String[] friend_person = line.split("\t");

            String friend=friend_person[0];//rita
            String[] persons=friend_person[1].split(",");

            for(int i=0;i<persons.length-1;i++){
                for(int j=i+1;j<persons.length;j++)
                { context.write(new Text(persons[i]+"和"+persons[j]),new Text(friend));
                }
            }

        }
    }

    public static class MyReduce2 extends Reducer<Text,Text,Text,Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {

            StringBuffer buf = new StringBuffer();
            Set<String> set = new HashSet<String>();
            for(Text str:values)
            {
                if(!set.contains(str.toString()))
                {
                    set.add(str.toString());
                }
            }
            for(String s:set){
                buf.append(s).append(",");
            }
            buf = buf.deleteCharAt(buf.length()-1);

            context.write(key,new Text("共同好友是:"+buf.toString()));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(StepTwo.class);

        job.setMapperClass(MyMapper2.class);
        job.setReducerClass(MyReduce2.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.setInputPaths(job,new Path(args[0]));
        Path path2 = new Path(args[1]);
        FileSystem fs = path2.getFileSystem(conf);
        if(fs.exists(path2)) {
            fs.delete(path2, true);
        }
        FileOutputFormat.setOutputPath(job,path2);
        System.exit( job.waitForCompletion(true)?0:1);
    }
}
